package com.me.springkafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.AcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

/**
 * @author zhengchubin
 * @create 2017-11-08 下午7:29
 **/
@Service
public class MsgAckConsumer implements AcknowledgingMessageListener<String,String> {

    @KafkaListener(topics = "hello")
    @Override
    public void onMessage(ConsumerRecord<String, String> stringStringConsumerRecord, Acknowledgment acknowledgment) {
        System.out.println("stringStringConsumerRecord:"+stringStringConsumerRecord);
//        if( 1==1){
//            throw new NullPointerException();
//        }
        acknowledgment.acknowledge();
    }
}
